// Decompiled by Jad v1.5.8g. Copyright 2001 Pavel Kouznetsov.
// Jad home page: http://www.kpdus.com/jad.html
// Decompiler options: packimports(3) 
// Source File Name:   AuthAgainstRedisCluster.java

package com.example.KafkaDdemo.auth;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import javax.security.auth.callback.*;
import javax.security.auth.login.AppConfigurationEntry;
import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
import org.apache.kafka.common.security.plain.PlainAuthenticateCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;

public class AuthAgainstRedisCluster
    implements AuthenticateCallbackHandler
{

    public AuthAgainstRedisCluster()
    {
        redisClusterPoolMaxTotal = 10;
        redisClusterPoolMaxIdle = 5;
        redisClusterPoolMaxWaitMillis = 30000L;
        redisClusterConnectionTimeoutMillis = 30000;
        redisClusterSoTimeoutMillis = 30000;
        redisClusterMaxAttempts = 1;
    }

    public void handle(Callback callbacks[])
        throws IOException, UnsupportedCallbackException
    {
        String username = null;
        Callback acallback[];
        int j = (acallback = callbacks).length;
        for(int i = 0; i < j; i++)
        {
            Callback callback = acallback[i];
            if(callback instanceof NameCallback)
                username = ((NameCallback)callback).getDefaultName();
            else
            if(callback instanceof PlainAuthenticateCallback)
            {
                PlainAuthenticateCallback plainCallback = (PlainAuthenticateCallback)callback;
                boolean authenticated = auth(username, plainCallback.password());
                plainCallback.authenticated(authenticated);
            } else
            {
                throw new UnsupportedCallbackException(callback);
            }
        }

    }

    private boolean auth(String username, char password[])
    {
        try
        {
            String inputPasswordMd5 = md5(String.valueOf(password));
            log.info("==== Authenticating username [{}] against md5 password [{}]", username, inputPasswordMd5);
            return inputPasswordMd5.equals(getPasswordFromRedisCluster((new StringBuilder(String.valueOf(redisClusterUserKeyPrefix))).append(username).toString()));
        }
        catch(Exception e)
        {
            return false;
        }
    }

    private String md5(String s)
    {
        try
        {
            MessageDigest md = MessageDigest.getInstance("MD5");
            md.update(s.getBytes(StandardCharsets.UTF_8));
            return Base64.getEncoder().encodeToString(md.digest());
        }
        catch(NoSuchAlgorithmException e)
        {
            log.error("==== Authentication failed due to the failure of MD5:", e);
        }
        return null;
    }

    public void configure(Map configs, String saslMechanism, List jaasConfigEntries)
    {
        redisClusterPassword = (String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.password");
        redisClusterUserKeyPrefix = (String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.userkey.prefix");
        redisClusterHosts = (String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.hosts");
        redisClusterPoolMaxTotal = Integer.parseInt((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.pool.maxtotal"));
        redisClusterPoolMaxIdle = Integer.parseInt((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.pool.maxidle"));
        redisClusterPoolMaxWaitMillis = Long.parseLong((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.pool.maxwaitmillis"));
        redisClusterConnectionTimeoutMillis = Integer.parseInt((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.connection.timeoutmillis"));
        redisClusterSoTimeoutMillis = Integer.parseInt((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.so.timeoutmillis"));
        redisClusterMaxAttempts = Integer.parseInt((String)((AppConfigurationEntry)jaasConfigEntries.get(0)).getOptions().get("redis.cluster.max.attempts"));
        log.info("==== Got redisClusterPassword MD5 = [{}], redisClusterUserKeyPrefix = [{}], redisClusterHosts = [{}], redisClusterPoolMaxTotal = [{}], redisClusterPoolMaxIdle = [{}], redisClusterPoolMaxWaitMillis = [{}], redisClusterConnectionTimeoutMillis = [{}], redisClusterSoTimeoutMillis = [{}], redisClusterMaxAttempts = [{}]", new Object[] {
            redisClusterPassword != null && !redisClusterPassword.trim().isEmpty() ? md5(redisClusterPassword) : "", redisClusterUserKeyPrefix, redisClusterHosts, Integer.valueOf(redisClusterPoolMaxTotal), Integer.valueOf(redisClusterPoolMaxIdle), Long.valueOf(redisClusterPoolMaxWaitMillis), Integer.valueOf(redisClusterConnectionTimeoutMillis), Integer.valueOf(redisClusterSoTimeoutMillis), Integer.valueOf(redisClusterMaxAttempts)
        });
    }

    private synchronized String getPasswordFromRedisCluster(String key)
    {
        try
        {
            return jedisCluster.get(key);
        }
        catch(Exception e) { }
        if(connectRedisCluster())
            return jedisCluster.get(key);
        else
            return null;
    }

    private boolean connectRedisCluster()
    {
        close();
        try
        {
            String hostports[] = redisClusterHosts.split(",");
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            jedisPoolConfig.setMaxTotal(redisClusterPoolMaxTotal);
            jedisPoolConfig.setMaxIdle(redisClusterPoolMaxIdle);
            jedisPoolConfig.setMaxWaitMillis(redisClusterPoolMaxWaitMillis);
            jedisPoolConfig.setTestOnBorrow(true);
            Set hostAndPortSet = new HashSet();
            String as[];
            int j = (as = hostports).length;
            for(int i = 0; i < j; i++)
            {
                String hp = as[i];
                String hostport[] = hp.split(":");
                hostAndPortSet.add(new HostAndPort(hostport[0], Integer.parseInt(hostport[1])));
            }

            if(redisClusterPassword != null && !redisClusterPassword.isEmpty())
            {
                jedisCluster = new JedisCluster(hostAndPortSet, redisClusterConnectionTimeoutMillis, redisClusterSoTimeoutMillis, redisClusterMaxAttempts, redisClusterPassword, jedisPoolConfig);
                log.info("==== Connected to redis cluster [{}] with password", redisClusterHosts);
            } else
            {
                jedisCluster = new JedisCluster(hostAndPortSet);
                log.info("==== Connected to redis cluster [{}] without password", redisClusterHosts);
            }
        }
        catch(Exception e)
        {
            log.error("==== Authentication failed due to the failure of connecting redis:", e);
            return false;
        }
        return true;
    }

    public void close()
    {
        try
        {
            jedisCluster.close();
        }
        catch(Exception exception) { }
        jedisCluster = null;
        log.info("==== Closed connection to redis");
    }

    private static final Logger log = LoggerFactory.getLogger(AuthAgainstRedisCluster.class);
    private String redisClusterPassword;
    private String redisClusterUserKeyPrefix;
    private JedisCluster jedisCluster;
    private String redisClusterHosts;
    private int redisClusterPoolMaxTotal;
    private int redisClusterPoolMaxIdle;
    private long redisClusterPoolMaxWaitMillis;
    private int redisClusterConnectionTimeoutMillis;
    private int redisClusterSoTimeoutMillis;
    private int redisClusterMaxAttempts;


    public static void main(String[] args) {
        AuthAgainstRedisCluster aarc = new AuthAgainstRedisCluster();
        aarc.redisClusterPassword = "123456";
        aarc.redisClusterUserKeyPrefix = "KAFA_USER_";
        aarc.redisClusterHosts = "10.92.219.80:6378,10.92.219.80:6379,10.92.219.81:6378,10.92.219.81:6379,10.92.219.84:6378,10.92.219.84:6379";
        aarc.connectRedisCluster();
        aarc.jedisCluster.set( aarc.redisClusterUserKeyPrefix + "admin", aarc.md5("12345678") );
        System.out.println( aarc.jedisCluster.get(aarc.redisClusterUserKeyPrefix + "admin") );

        aarc.jedisCluster.set(aarc.redisClusterUserKeyPrefix + "bob", aarc.md5("bobpassword") );
        System.out.println( aarc.jedisCluster.get(aarc.redisClusterUserKeyPrefix + "bob") );

    }

}
